package com.bkd.report

import java.util.Properties

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object AreaRpt {

  def main(args: Array[String]): Unit = {
    if(args.length !=1){
      println(
        """
          |com.bkd.report.AreaRpt
          |参数
          |logInputPath
        """.stripMargin)
      sys.exit()
    }

    //制定接受的参数
    val Array(logInPutPath) = args
    //创建sparkConf
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.setMaster("local[*]")
    //制定序列化方式
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    //创建SparkContext
    val sc = new SparkContext(conf)
    //获得sqlContext
    val sqlContext = new SQLContext(sc)
    //读取文件
    val parquetData: DataFrame = sqlContext.read.parquet(logInPutPath)
    //注册表
    parquetData.registerTempTable("log")

    //使用sparkSql执行业务逻辑
    val result = sqlContext.sql(
      """|select
        |provincename,
        |cityname,
        |sum(case when requestmode=1 and processnode >=1 then 1 else 0 end) ysqq,
        |sum(case when requestmode=1 and processnode >=2 then 1 else 0 end) yxqq,
        |sum(case when requestmode=1 and processnode >=3 then 1 else 0 end) ggqq,
        |
        |sum(case when iseffective=1 and isbilling=1 and isbid=1 and adorderid !=0 then 1 else 0 end) cyjjs,
        |sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) jjcgs,
        |sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) zsh,
        |sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) djs,
        |sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1.0*winprice/1000 else 0 end) ggcb,
        |sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1.0*adpayment/1000 else 0 end) ggxf
        |from log
        |group by provincename,
         |
         |cityname
      """.stripMargin)
    val load: Config = ConfigFactory.load()
    val props = new Properties()
    props.setProperty("user",load.getString("jdbc.user"))
    props.setProperty("password",load.getString("jdbc.password"))
    result.write.jdbc(load.getString("jdbc.url"),load.getString("jdbc.tableName"),props)
    sc.stop()

  }
}
